package dm.ass2.eval;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;

import dm.ass2.eval.io.*;
import dm.ass2.io.CodebookUsersWritable;

public class EvalReducer extends Reducer<IntWritable , ErrorWritable, Text, Text> {
	private static final Log LOG = LogFactory.getLog(EvalReducer.class);
	
	@Override
	protected void reduce(IntWritable key, Iterable<ErrorWritable> errorItems, Context context) throws IOException, InterruptedException {
	  double error = 0;
	  int count = 0;
	  for (ErrorWritable obj : errorItems){
	       for (int i = 0 ; i <  obj.getSize() ;i++){
               error = error + obj.getError()[i];
               count = count + obj.getNum()[i];
	  }
	  }
	  double rmse = Math.pow(error/count,0.5);
	  LOG.info(String.format("the final result is: " + Double.toString(rmse) , Double.toString(rmse)));
	  Text keyOut   = new Text("1");
	  Text valueOut = new Text(Double.toString(rmse)); 
	  context.write(keyOut, valueOut);
	}
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
	}
	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		super.cleanup(context);
	}
	}
	